using System;
using System.Security.Principal;
using Common.Logging;
using SimpleServiceBus.Bus.MessageManagement;
using SimpleServiceBus.Bus.MessageManagement.Callbacks;
using SimpleServiceBus.Bus.Pipeline;
using SimpleServiceBus.Bus.Subscriptions;
using SimpleServiceBus.Bus.Transport;
using SimpleServiceBus.Endpoint;
using SimpleServiceBus.Messages.ControlMessages;
using SimpleServiceBus.Messages.Subscriptions;

namespace SimpleServiceBus.Bus
{
    /// <summary>
    /// A unicast implementation of <see cref="IMessageBus"/> for SimpleServiceBus.
    /// </summary>
    public class MessageBus : EndpointServiceBase,IDistributorWorker
    {
        public event EventHandler<MessageProcessingErrorEventArgs> MessageProcessingException;

        public event EventHandler<MessageProcessingEventArgs> BeginReceiveMessage;
        public event EventHandler<MessageProcessingEventArgs> MessageReceived;
        public event EventHandler<MessageProcessingEventArgs> BeginSendMessage;
        public event EventHandler<MessageProcessingEventArgs> MessageSent;

        private static readonly ILog _log = LogManager.GetLogger(typeof(MessageBus));

        private volatile bool _canSendReadyMessages = true;

        public virtual ITransport Transport
        {
            get; private set;
        }

        /// <summary>
        /// Should be used by programmer, not administrator.
        /// Sets whether or not the return address of a received message 
        /// should be propogated when the message is forwarded. This field is
        /// used primarily for the Distributor.
        /// </summary>
        public virtual bool PropogateReturnAddressOnSend { get; set; }

       /// <summary>
        /// Should be used by administrator, not programmer.
        /// Sets the address to which the messages received on this bus
        /// will be sent when the method HandleCurrentMessageLater is called.
        /// </summary>
        public virtual string DistributorDataAddress { get; set; }

        /// <summary>
        /// Should be used by administrator, not programmer.
        /// Sets the address of the distributor control queue.
        /// </summary>
        /// <remarks>
        /// Notifies the given distributor
        /// when a thread is now available to handle a new message.
        /// </remarks>
        public virtual string DistributorControlAddress { get; set; }

        public virtual MessageEnvelope MessageCurrentlyBeingReceived
        {
            get
            {
                if (MessagePipeline.CurrentReceivePipeline != null)
                    return MessagePipeline.CurrentReceivePipeline.CurrentContext.MessageEnvelope;
                return null;
            }
        }

        public virtual MessageEnvelope MessageCurrentlyBeingSent
        {
            get
            {
                if (MessagePipeline.CurrentSendPipeline != null)
                    return MessagePipeline.CurrentSendPipeline.CurrentContext.MessageEnvelope;
                return null;
            }
        }

        public virtual string ReturnAddress
        {
            get { return Transport.ReturnAddress; }
        }

        public void StopSendingReadyMessages()
        {
            _canSendReadyMessages = false;
        }

        public void ContinueSendingReadyMessages()
        {
            _canSendReadyMessages = true;
        }

        /// <summary>
        /// Publishes the first message in the list to all subscribers of that message type.
        /// </summary>
        /// <param name="message">A list of messages.  Only the first will be published.</param>
        public virtual void Publish(object message)
        {
            var subManager = ServiceManager.GetService<ISubscriptionManager>();
            if (subManager != null)
            {
                subManager.Publish(message);
            }
        }
        
        public virtual void Subscribe<T>() 
        {
            var msgMgr = ServiceManager.GetService<IMessageTypeManagementService>();
            if (msgMgr != null)
            {
                string destination = msgMgr.GetDestinationFor<T>();
                if (!String.IsNullOrEmpty(destination))
                {
                    var request = new SubscribeToMessageRequest
                                      {
                                          MessageType = typeof(T).AssemblyQualifiedName,
                                          SubscriberAddress = Transport.ReturnAddress
                                      };
                    Send(request, destination);
                }
                else
                    throw new Exception(
                        "MessageEnvelope Subscription failed: no destination address was found for the message type " +
                        typeof (T).Name);
            }
        }

        /// <summary>
        /// Unsubscribes from receiving published messages of the specified type.
        /// </summary>
        public virtual void Unsubscribe<T>()
        {
            var msgMgr = ServiceManager.GetService<IMessageTypeManagementService>();
            if (msgMgr != null)
            {
                string destination = msgMgr.GetDestinationFor<T>();
                if (!String.IsNullOrEmpty(destination))
                {
                    var request = new UnsubscribeFromMessageRequest
                    {
                        MessageType = typeof(T).AssemblyQualifiedName,
                        SubscriberAddress = Transport.ReturnAddress
                    };
                    Send(request, destination);
                }
				 else
                    throw new Exception(
                        "MessageEnvelope Unsubscribe failed: no destination address was found for the message type " +
                        typeof (T).Name);
            }
        }
		
		 public virtual ICallback Send(object message)
        {
            var mManager = ServiceManager.GetService<IMessageTypeManagementService>();
            if (mManager != null)
            {
                string destination = mManager.GetDestinationForMessage(message);
                if (!string.IsNullOrEmpty(destination))
                    return Send(message, destination);
            }
            throw new Exception(("A destination could not be located for this type of message:" + message.GetType().Name));
        }

       
        public virtual ICallback Send(object message,params string[] destinations)
        {
            try
            {
                MessageEnvelope toSend;

                if (message is MessageEnvelope)
                    toSend = (MessageEnvelope) message;
                else
                    toSend = GetMessageEnvelopeFor(message, destinations);
                
                ICallback callback = null;
                var cbm = ServiceManager.GetService<ICorrelatedCallbackManager>();

                if (cbm != null)
                    callback = cbm.Register(toSend.MessageID);
                
                var factory = ServiceManager.GetService<IMessagePipelineFactory>();
                if (factory != null)
                {
                    IMessagePipeline pipeline = factory.CreatePipelineFor(toSend, PipelineDirection.Send);
                    if (pipeline != null)
                    {
                        OnBeginSendMessage(toSend, pipeline);

                        pipeline.ProcessMessage();
                        if (pipeline.CurrentContext.AbortMessageProcessing)
                        {
                            if (cbm != null)
                                cbm.Unregister(toSend.MessageID);
                            return null;
                        }
                        
                    }
                }

                Transport.Send(toSend, destinations);

                OnMessageSent(toSend);

                return callback;
            }
            catch (Exception ex)
            {
                _log.Error("Error sending message", ex);
                throw;
            }
        }
		
		 
        public void SendLocal(object message)
        {
            //if we're a worker, send to the distributor data bus
            if (DistributorDataAddress != null)
            {
                Send(message, DistributorDataAddress);
            }
            else
            {
                Send(message, Transport.ReturnAddress);
            }
        }       
        
        public void Reply(object message)
        {
            MessageEnvelope currentMessageEnvelope = MessageCurrentlyBeingReceived;
            
            if (string.IsNullOrEmpty(currentMessageEnvelope.ReturnAddress))
                throw new InvalidOperationException(
                    "The message being replied to does not have a return address configured. It was probably sent from a Send Only endpoint, and cannot receive messages.");
            
            MessageEnvelope toSend = GetMessageEnvelopeFor(message,currentMessageEnvelope.ReturnAddress);

            toSend.CorrelationID = currentMessageEnvelope.MessageID;

            Send(toSend, currentMessageEnvelope.ReturnAddress);

            var cbm = ServiceManager.GetService<ICorrelatedCallbackManager>();
            if (cbm != null)
                // Unregister since we don't need to wait on any response (to the reply)
                cbm.Unregister(toSend.MessageID);
        }
       

        public void Forward(MessageEnvelope messageEnvelope, params string[] destinations)
        {
            Transport.Send(messageEnvelope, destinations);
        }
       


    
        /// <summary>
        /// If this bus is configured to feed off of a distributor,
        /// it will send a <see cref="ReadyMessage"/> to its control address.
        /// </summary>
        /// <param name="startup"></param>
        protected virtual void SendReadyMessage(bool startup)
        {
            if (String.IsNullOrEmpty(DistributorControlAddress ))
                return;

            if (String.IsNullOrEmpty(Transport.ReturnAddress)) 
                return;


            if (!_canSendReadyMessages)
                return;

            var rm = new ReadyMessage
                         {
                             NumberOfWorkerThreads = Transport.NumberOfWorkerThreads
                         };

            if (startup)
                rm.ClearPreviousFromThisAddress = true;

            MessageEnvelope toSend = GetMessageEnvelopeFor(rm, DistributorControlAddress);
            toSend.ReturnAddress = Transport.ReturnAddress;

            Send(toSend, DistributorControlAddress);

            _log.Debug("Sending ReadyMessage to " + DistributorControlAddress);
        }


        protected override void ShutDownService()
        {
            //Transport.Stop();
            Transport.MessageProcessingError -= HandleMessageProcessingError;
            Transport.MessageReceived -= AcceptMessageFromTransport;
        }

        protected override void StartService()
        {
            AppDomain.CurrentDomain.SetPrincipalPolicy(PrincipalPolicy.WindowsPrincipal);

            Transport = ServiceManager.GetService<ITransport>();
            if (Transport == null)
                throw new Exception(
                    "No transport has been configured for this endpoint. Every endpoint must have a valid transport registered in the LocalServiceManager.");

            Transport.MessageProcessingError += HandleMessageProcessingError;
            Transport.MessageReceived += AcceptMessageFromTransport;
            
        }

        void HandleMessageProcessingError(object sender, MessageProcessingErrorEventArgs e)
        {
            if (e.Exception is MessageProcessingException)
            {
                var error = (MessageProcessingException)e.Exception;
                var errorHandler =
                    Endpoint.LocalServices.GetService<IMessageProcessingErrorHandler>();

                if (errorHandler != null && errorHandler.HandleMessageProcessingError(ref error,e.Message,this) == false)
                    return;

                //In case the handler replaced or wrapped the original exception, we'll 
                //replace e.Exception with the possibly new value
                if (error != null)
                    e.Exception = error;
                
            }

            if (!OnUnhandledException(e.Exception))
            {
                _log.Error("Unhandled messaging exception.", e.Exception);
                throw e.Exception;
            }
        }

        protected virtual bool OnUnhandledException(Exception exception)
        {
            EventHandler<MessageProcessingErrorEventArgs> evt = MessageProcessingException;
            if (evt != null)
            {
                evt(this, new MessageProcessingErrorEventArgs {Exception = exception});
                return true;
            }
            return false;
        }

        protected virtual void OnBeginSendMessage(MessageEnvelope message,IMessagePipeline pipeline)
        {
            EventHandler<MessageProcessingEventArgs> evt = BeginSendMessage;
            if (evt != null)
                evt(this, new MessagePipelineEventArgs {Pipeline = pipeline, Message = message});

        }

        protected virtual void OnMessageSent(MessageEnvelope message)
        {
            EventHandler<MessageProcessingEventArgs> evt = MessageSent;
            if (evt != null)
                evt(this, new MessageProcessingEventArgs {Message = message});
        }

        protected virtual void OnBeginReceiveMessage(MessageEnvelope message, IMessagePipeline pipeline)
        {
            EventHandler<MessageProcessingEventArgs> evt = BeginReceiveMessage;
            if (evt != null)
                evt(this, new MessagePipelineEventArgs{Pipeline = pipeline,Message = message});
        }

        protected virtual void OnMessageReceived(MessageEnvelope message)
        {
            EventHandler<MessageProcessingEventArgs> evt = MessageReceived;
            if (evt != null)
                evt(this, new MessageProcessingEventArgs {Message = message});
        }


        #region receiving and handling


        protected virtual void HandleMessage(MessageEnvelope m)
        {
            try
            {
                var plFactory = ServiceManager.GetService<IMessagePipelineFactory>();
                if (plFactory == null)
                    throw new Exception("No PipelineFactory has been configured for this endpoint.");

                IMessagePipeline pipeline = plFactory.CreatePipelineFor(m, PipelineDirection.Receive);

                OnBeginReceiveMessage(m,pipeline);

                pipeline.PipelineEvents.PipelineError += Events_PipelineError;
                pipeline.ProcessMessage();

                OnMessageReceived(m);
            }
            catch (Exception ex)
            {
                _log.Error("Unhandled error processing message", ex);
                throw;
            }
        }

       protected virtual void AcceptMessageFromTransport(object sender, MessageReceivedEventArgs e)
        {
            MessageEnvelope msg = e.MessageEnvelope;

            if (msg.IsInitializationMessageFor(this))
            {
                _log.Info(Transport.ReturnAddress + " initialized.");
                return;
            }

            //if (MessageReceived != null)
            //    MessageReceived(msg);

            HandleMessage(msg);

            SendReadyMessage(false);

            _log.Debug("Finished handling message.");
        }


        #endregion

        #region helper methods

        /// <summary>
        /// Wraps the provided messages in an SimpleServiceBus envelope.
        /// </summary>
        /// <param name="destinations">The destination to which to send the messages.</param>
        /// <param name="message">The messages to wrap.</param>
        /// <returns>The envelope containing the messages.</returns>
        protected virtual MessageEnvelope GetMessageEnvelopeFor(object message, params string[] destinations)
        {
            var result = new MessageEnvelope
                             {
                                 MessageID = GetNextMessageID(),
                                 Body = message,
                                 ReturnAddress = GetReturnAddressFor(destinations)
                             };            

            TimeSpan timeToBeReceived = TimeSpan.MaxValue;

            result.Recoverable = RecoverableAttribute.IsRecoverable(message);

            foreach (
                TimeToBeReceivedAttribute a in
                    message.GetType().GetCustomAttributes(typeof(TimeToBeReceivedAttribute), true))
                timeToBeReceived = (a.TimeToBeReceived < timeToBeReceived ? a.TimeToBeReceived : timeToBeReceived);


            result.TimeToBeReceived = timeToBeReceived;

            return result;
        }

        protected virtual string GetNextMessageID()
        {
            var generator = ServiceManager.GetService<IMessageIdGenerator>();
            if (generator == null)
                return Guid.NewGuid().ToString();
            return generator.GenerateMessageID();
        }

        protected virtual string GetReturnAddressFor(params string[] destinations)
        {
            string result = Transport.ReturnAddress;
			
            if (PropogateReturnAddressOnSend)
				return result;

            // if we're a worker
            if (DistributorDataAddress != null)
            {
                result = DistributorDataAddress;

                //if we're sending a message to the control bus, then use our own address
                if (destinations != null &&
                    destinations.Length > 0 &&
                    destinations[0] == DistributorControlAddress)
                    result = Transport.ReturnAddress;
            }			

            return result;
        }

        #endregion

        protected void Events_PipelineError(object sender, MessagePipelineErrorEventArgs e)
        {
            HandleMessageProcessingError(this,
                                         new MessageProcessingErrorEventArgs
                                             {Exception = e.Error, Message = e.Pipeline.CurrentContext.MessageEnvelope});
        }
    }
}